package org.jpush.core;

import org.jpush.core.AbstractChannelManager.ChannelContext;
import org.jpush.core.AbstractChannelManager.ChannelInfo;
import org.jpush.core.Channel.PacketInfo;
import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.Pointer;

public class ChannelStreamer extends AbstractChannelStreamer
{
	protected ChannelManager channelManager;
	
	public ChannelStreamer(ChannelManager channelManager)
	{
		this.channelManager = channelManager;
	}

	protected ProcessResult processConnection(LogicalConnection logicalConnection)
	{
	    if (PhysicalConnection.notConnected(logicalConnection.physicalConnection.getStatus()))
	    {
	        logicalConnection.decrementUsage();
	        return ProcessResult.DeleteItem;
	    }

	    //Try to send one packet :
	    Pointer<ChannelInfo> pFoundInfo = new Pointer<ChannelInfo>();
	    Pointer<PacketInfo> pPacketInfo = new Pointer<PacketInfo>();

	    ChannelContext channelCtxt = logicalConnection.physicalConnection.getChannelContext();
	    String subscriberKey = logicalConnection.getKey();

	    OutgoingPacket packet = channelManager.getNextPacket(subscriberKey, channelCtxt, pPacketInfo, pFoundInfo);
	    if (packet == null)
	        return ProcessResult.PauseProcessing;

	    SendResult result = logicalConnection.tryPushPacket(packet);

	    channelManager.disposePacket(pPacketInfo.getValue(), subscriberKey, pFoundInfo.getValue(), channelCtxt, result == SendResult.SendResult_OK);

	    return ProcessResult.ContinueProcessing;
	}
	

	public void addItem(LogicalConnection item)
	{
	    LogicalConnection logicalConnection = (LogicalConnection) item;
	    logicalConnection.incrementUsage();

	    super.addItem(item);
	}


}
